#!/usr/bin/env python3
"""
Rubberband MCP Server

面向 rubberband 音频时间拉伸和音高变换命令行工具

MCP tools 列表：
1. rubberband_time_stretch - 执行时间拉伸操作
   - 参数：input_file, output_file, time_ratio, duration, pitch_ratio
   - 返回：命令执行结果

2. rubberband_pitch_shift - 执行音高变换操作
   - 参数：input_file, output_file, pitch_shift, frequency_ratio
   - 返回：命令执行结果

3. rubberband_crispness - 设置处理清晰度
   - 参数：input_file, output_file, crispness
   - 返回：命令执行结果

4. rubberband_formant - 控制共振峰保持
   - 参数：input_file, output_file, preserve_formants
   - 返回：命令执行结果

5. rubberband_threading - 控制线程使用
   - 参数：input_file, output_file, no_threads
   - 返回：命令执行结果

6. rubberband_advanced - 高级处理选项
   - 参数：input_file, output_file, window_short, detector_soft
   - 返回：命令执行结果
"""

import subprocess
import shlex
from typing import Optional
from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("Rubberband MCP Server")

@mcp.tool()
def rubberband_time_stretch(
    input_file: str,
    output_file: str,
    time_ratio: Optional[float] = None,
    duration: Optional[float] = None,
    pitch_ratio: Optional[float] = None
) -> dict:
    """
    执行 rubberband 时间拉伸操作
    
    参数:
        input_file: 输入音频文件路径
        output_file: 输出音频文件路径
        time_ratio: 时间拉伸比例（如 2.0 表示两倍速，0.5 表示半速）
        duration: 目标时长（秒）
        pitch_ratio: 音高比例（可选）
    
    返回:
        {
            "success": bool,    # 命令是否成功执行
            "command": str,     # 执行的完整命令
            "exit_code": int,   # 退出码
            "stdout": str,      # 标准输出
            "stderr": str       # 标准错误
        }
    """
    cmd_parts = ["rubberband"]
    
    if time_ratio is not None:
        cmd_parts.append(f"-t{time_ratio}")
    
    if duration is not None:
        cmd_parts.append(f"-D{duration}")
    
    if pitch_ratio is not None:
        cmd_parts.append(f"-T{pitch_ratio}")
    
    cmd_parts.extend([input_file, output_file])
    command = " ".join(cmd_parts)
    
    try:
        result = subprocess.run(
            shlex.split(command),
            capture_output=True,
            text=True,
            timeout=30
        )
        return {
            "success": result.returncode == 0,
            "command": command,
            "exit_code": result.returncode,
            "stdout": result.stdout,
            "stderr": result.stderr
        }
    except Exception as e:
        return {
            "success": False,
            "command": command,
            "exit_code": -1,
            "stdout": "",
            "stderr": str(e)
        }

@mcp.tool()
def rubberband_pitch_shift(
    input_file: str,
    output_file: str,
    pitch_shift: Optional[int] = None,
    frequency_ratio: Optional[float] = None
) -> dict:
    """
    执行 rubberband 音高变换操作
    
    参数:
        input_file: 输入音频文件路径
        output_file: 输出音频文件路径
        pitch_shift: 音高移动（半音数，如 2 表示升高两个半音）
        frequency_ratio: 频率比例（如 1.5 表示频率变为1.5倍）
    
    返回:
        统一 JSON 结构
    """
    cmd_parts = ["rubberband"]
    
    if pitch_shift is not None:
        cmd_parts.append(f"-p{pitch_shift}")
    
    if frequency_ratio is not None:
        cmd_parts.append(f"-f{frequency_ratio}")
    
    cmd_parts.extend([input_file, output_file])
    command = " ".join(cmd_parts)
    
    try:
        result = subprocess.run(
            shlex.split(command),
            capture_output=True,
            text=True,
            timeout=30
        )
        return {
            "success": result.returncode == 0,
            "command": command,
            "exit_code": result.returncode,
            "stdout": result.stdout,
            "stderr": result.stderr
        }
    except Exception as e:
        return {
            "success": False,
            "command": command,
            "exit_code": -1,
            "stdout": "",
            "stderr": str(e)
        }

@mcp.tool()
def rubberband_crispness(
    input_file: str,
    output_file: str,
    crispness: int
) -> dict:
    """
    设置 rubberband 处理清晰度
    
    参数:
        input_file: 输入音频文件路径
        output_file: 输出音频文件路径
        crispness: 清晰度级别 (0-6)
    
    返回:
        统一 JSON 结构
    """
    command = f"rubberband -c{crispness} {input_file} {output_file}"
    
    try:
        result = subprocess.run(
            shlex.split(command),
            capture_output=True,
            text=True,
            timeout=30
        )
        return {
            "success": result.returncode == 0,
            "command": command,
            "exit_code": result.returncode,
            "stdout": result.stdout,
            "stderr": result.stderr
        }
    except Exception as e:
        return {
            "success": False,
            "command": command,
            "exit_code": -1,
            "stdout": "",
            "stderr": str(e)
        }

@mcp.tool()
def rubberband_formant(
    input_file: str,
    output_file: str,
    preserve_formants: bool = False
) -> dict:
    """
    控制 rubberband 共振峰保持
    
    参数:
        input_file: 输入音频文件路径
        output_file: 输出音频文件路径
        preserve_formants: 是否保持共振峰
    
    返回:
        统一 JSON 结构
    """
    cmd_parts = ["rubberband"]
    
    if preserve_formants:
        cmd_parts.append("-L")
    
    cmd_parts.extend([input_file, output_file])
    command = " ".join(cmd_parts)
    
    try:
        result = subprocess.run(
            shlex.split(command),
            capture_output=True,
            text=True,
            timeout=30
        )
        return {
            "success": result.returncode == 0,
            "command": command,
            "exit_code": result.returncode,
            "stdout": result.stdout,
            "stderr": result.stderr
        }
    except Exception as e:
        return {
            "success": False,
            "command": command,
            "exit_code": -1,
            "stdout": "",
            "stderr": str(e)
        }

@mcp.tool()
def rubberband_threading(
    input_file: str,
    output_file: str,
    no_threads: bool = False
) -> dict:
    """
    控制 rubberband 线程使用
    
    参数:
        input_file: 输入音频文件路径
        output_file: 输出音频文件路径
        no_threads: 是否禁用多线程
    
    返回:
        统一 JSON 结构
    """
    cmd_parts = ["rubberband"]
    
    if no_threads:
        cmd_parts.append("--no-threads")
    
    cmd_parts.extend([input_file, output_file])
    command = " ".join(cmd_parts)
    
    try:
        result = subprocess.run(
            shlex.split(command),
            capture_output=True,
            text=True,
            timeout=30
        )
        return {
            "success": result.returncode == 0,
            "command": command,
            "exit_code": result.returncode,
            "stdout": result.stdout,
            "stderr": result.stderr
        }
    except Exception as e:
        return {
            "success": False,
            "command": command,
            "exit_code": -1,
            "stdout": "",
            "stderr": str(e)
        }

@mcp.tool()
def rubberband_advanced(
    input_file: str,
    output_file: str,
    window_short: bool = False,
    detector_soft: bool = False
) -> dict:
    """
    rubberband 高级处理选项
    
    参数:
        input_file: 输入音频文件路径
        output_file: 输出音频文件路径
        window_short: 是否使用短窗口
        detector_soft: 是否使用软检测器
    
    返回:
        统一 JSON 结构
    """
    cmd_parts = ["rubberband"]
    
    if window_short:
        cmd_parts.append("--window-short")
    
    if detector_soft:
        cmd_parts.append("--detector-soft")
    
    cmd_parts.extend([input_file, output_file])
    command = " ".join(cmd_parts)
    
    try:
        result = subprocess.run(
            shlex.split(command),
            capture_output=True,
            text=True,
            timeout=30
        )
        return {
            "success": result.returncode == 0,
            "command": command,
            "exit_code": result.returncode,
            "stdout": result.stdout,
            "stderr": result.stderr
        }
    except Exception as e:
        return {
            "success": False,
            "command": command,
            "exit_code": -1,
            "stdout": "",
            "stderr": str(e)
        }

if __name__ == "__main__":
    mcp.run()